package org.alex.mqtt.common;

import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.timeout.IdleStateHandler;
import org.alex.mqtt.common.bus.*;
import org.alex.mqtt.common.protocol.MqttProcessor;
import org.alex.mqtt.common.retain.RetainMessageHandler;
import org.alex.mqtt.common.support.config.BrokerConfig;
import org.alex.mqtt.common.support.helper.Pair;
import org.alex.mqtt.common.support.log.AlexLogger;
import org.alex.mqtt.common.session.MQTTSession;
import org.slf4j.Logger;

import java.util.Map;
import java.util.concurrent.ExecutorService;

/**
 * 处理mqtt连接并解析mqtt协议
 *
 * @author : alex
 */
public class MQTTConnection {
    private static final Logger log = AlexLogger.mqttLog;

    /**
     * mqtt通道
     */
    private Channel channel;

    private Map<Integer, Pair<MqttProcessor, ExecutorService>> processorTable;

    /**
     * mqtt broker相关配置
     */
    private BrokerConfig brokerConfig;

    /**
     * mqtt会话
     */
    private MQTTSession mqttSession;

    /**
     * 认证
     */
    private Authenticator authenticator;

    /**
     * session管理器
     */
    private DeviceSessionManager deviceSessionManager;

    /**
     * 订阅管理器
     */
    private DeviceSubscriptionManager deviceSubscriptionManager;

    /**
     * 消息管理器
     */
    private DeviceMessageManager deviceMessageManager;

    /**
     * 保存消息处理器
     */
    private RetainMessageHandler retainMessageHandler;

    /**
     * 集群事件管理器
     */
    private ClusterEventManager clusterEventManager;

    /**
     * 客户端id
     */
    private String clientId;

    /**
     * clientId 校验
     */
    public boolean clientIdVerify(String clientId) {
        return this.authenticator.clientIdVerify(clientId);
    }

    /**
     * 是否在黑名单
     */
    public boolean onBlackList(String remoteAddr, String clientId) {
        return this.authenticator.onBlackList(clientId, remoteAddr);
    }

    /**
     * 登录
     */
    public boolean login(String clientId, String username, byte[] password) {
        return this.authenticator.login(clientId, username, password);
    }

    /**
     * 心跳设置
     */
    public boolean keepAlive(int heartbeatSec) {
        int keepAlive = (int) (heartbeatSec * 1.5f);
        if (this.channel.pipeline().names().contains("idleStateHandler")) {
            this.channel.pipeline().remove("idleStateHandler");
        }
        this.channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(keepAlive, 0, 0));
        return true;
    }

    /**
     * 创建或者重新打开Session
     */
    public boolean createOrReopenSession(MqttConnectMessage connectMessage) {
        return false;
    }
}
